load libraries
In [1]:
import os
import cv2
import glob
import numpy as np
import pandas as pd
from keras.models import *
from keras.optimizers import *
from keras.layers import *
from keras.applications import *
from keras.preprocessing.image import *
In [2]:
dir = "/ext/Data/distracted_driver_detection/"
model_image_size = (320, 480)
fine_tune_layer = 116
final_layer = 135
visual_layer = 132
batch_size = 64
加载数据集¶
load train data
In [3]:
train_gen = ImageDataGenerator(
featurewise_std_normalization=True,
samplewise_std_normalization=False,
rotation_range=10.,
width_shift_range=0.05,
height_shift_range=0.05,
shear_range=0.1,
zoom_range=0.1,
)
gen = ImageDataGenerator(
featurewise_std_normalization=True,
samplewise_std_normalization=False,
)
train_generator = train_gen.flow_from_directory(os.path.join(dir, 'train'), model_image_size, shuffle=True, batch_size=batch_size, class_mode="categorical")
print("subdior to train type {}".format(train_generator.class_indices))
valid_generator = gen.flow_from_directory(os.path.join(dir, 'valid'), model_image_size, shuffle=True, batch_size=batch_size, class_mode="categorical")
print("subdior to valid type {}".format(valid_generator.class_indices))
In [4]:
input_tensor = Input((*model_image_size, 3))
x = input_tensor
x = Lambda(xception.preprocess_input)(x)
base_model = Xception(input_tensor=x, weights='imagenet', include_top=False)
x = GlobalAveragePooling2D()(base_model.output)
x = Dropout(0.5)(x)
x = Dense(10, activation='softmax')(x)
model = Model(base_model.input, x)
print("total layer count {}".format(len(base_model.layers)))
for i in range(fine_tune_layer):
model.layers[i].trainable = False
训练模型¶
In [5]:
print("train_generator.samples = {}".format(train_generator.samples))
print("valid_generator.samples = {}".format(valid_generator.samples))
steps_train_sample = train_generator.samples // 128 + 1
steps_valid_sample = valid_generator.samples // 128 + 1
In [6]:
model.compile(optimizer='adam', loss='categorical_crossentropy', metrics=['accuracy'])
model.fit_generator(train_generator, steps_per_epoch=steps_train_sample, epochs=4, validation_data=valid_generator, validation_steps=steps_valid_sample)
model.save("models/xception-imagenet-finetune{}-adam.h5".format(fine_tune_layer))
print("model saved!")
In [7]:
model.compile(optimizer=RMSprop(lr=1*0.00001), loss='categorical_crossentropy', metrics=['accuracy'])
model.fit_generator(train_generator, steps_per_epoch=steps_train_sample, epochs=6, validation_data=valid_generator, validation_steps=steps_valid_sample)
model.save("models/xception-imagenet-finetune{}.h5".format(fine_tune_layer))
print("model saved!")
In [8]:
from IPython.display import SVG
from keras.utils.vis_utils import model_to_dot
from keras.models import *
model = load_model("models/xception-imagenet-finetune{}.h5".format(fine_tune_layer))
print("load successed")
#SVG(model_to_dot(model).create(prog='dot', format='svg'))
CAM 可视化¶
http://cnnlocalization.csail.mit.edu/
$cam = (P-0.5)*w*output$
- cam: 类激活图 X*X
- P: 概率
- output: 卷积层的输出 2048*1
- w: 卷积核的权重 X*X*2048
In [9]:
z = zip([x.name for x in model.layers], range(len(model.layers)))
for k, v in z:
print("{} - {}".format(k,v))
In [10]:
import matplotlib.pyplot as plt
import random
%matplotlib inline
%config InlineBackend.figure_format = 'retina'
def show_heatmap_image(model_show, weights_show):
test_dir = os.path.join(dir, "test", "test" )
image_files = glob.glob(os.path.join(test_dir,"*"))
print(len(image_files))
plt.figure(figsize=(12, 24))
for i in range(10):
plt.subplot(5, 2, i+1)
img = cv2.imread(image_files[2000*i+113])
img = cv2.resize(img, (model_image_size[1],model_image_size[0]))
x = img.copy()
x.astype(np.float32)
out, predictions = model_show.predict(np.expand_dims(x, axis=0))
predictions = predictions[0]
out = out[0]
max_idx = np.argmax(predictions)
prediction = predictions[max_idx]
status = ["safe driving", " texting - right", "phone - right", "texting - left", "phone - left",
"operation radio", "drinking", "reaching behind", "hair and makeup", "talking"]
plt.title('c%d |%s| %.2f%%' % (max_idx , status[max_idx], prediction*100))
cam = (prediction - 0.5) * np.matmul(out, weights_show)
cam = cam[:,:,max_idx]
cam -= cam.min()
cam /= cam.max()
cam -= 0.2
cam /= 0.8
cam = cv2.resize(cam, (model_image_size[1],model_image_size[0]))
heatmap = cv2.applyColorMap(np.uint8(255*cam), cv2.COLORMAP_JET)
heatmap[np.where(cam <= 0.2)] = 0
out = cv2.addWeighted(img, 0.8, heatmap, 0.4, 0)
plt.axis('off')
plt.imshow(out[:,:,::-1])
print("done")
In [11]:
weights = model.layers[final_layer].get_weights()[0]
layer_output = model.layers[visual_layer].output
model2 = Model(model.input, [layer_output, model.output])
print("layer_output {0}".format(layer_output))
print("weights shape {0}".format(weights.shape))
show_heatmap_image(model2, weights)
In [12]:
def gen_kaggle_csv(model, model_image_size, csv_name):
dir = "/ext/Data/distracted_driver_detection/"
gen = ImageDataGenerator()
test_generator = gen.flow_from_directory(dir + "test/", model_image_size, shuffle=False,
batch_size=batch_size, class_mode=None)
# s = test_generator.__dict__
# del s['filenames']
# print(s)
y_pred = model.predict_generator(test_generator, steps=test_generator.samples//batch_size+1, verbose=1)
print("y_pred shape {}".format(y_pred.shape))
y_pred = y_pred.clip(min=0.005, max=0.995)
print(y_pred[:3])
l = list()
for i, fname in enumerate(test_generator.filenames):
name = fname[fname.rfind('/')+1:]
l.append( [name, *y_pred[i]] )
l = np.array(l)
data = {'img': l[:,0]}
for i in range(10):
data["c%d"%i] = l[:,i+1]
df = pd.DataFrame(data, columns=['img'] + ['c%d'%i for i in range(10)])
df.head(10)
df = df.sort_values(by='img')
df.to_csv(csv_name, index=None, float_format='%.3f')
print("csv saved")
print("done")
In [13]:
gen_kaggle_csv(model, model_image_size, 'csv/Xception-imagenet-finetune{}-pred.csv'.format(fine_tune_layer))
In [ ]: